package cn.ai.boot.config;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author jie.zhao@dms365.com
 * @Date 2025/7/2 11:20
 * 通用异步处理
 */
public class AsyncServiceUtils {
    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncService.class);

    /**
     * 异步执行任务 每次处理一批数据
     */
    public <T> void executeConcurrentTasks(ExecutorService threadPoolService,
                                           Collection<List<T>> list,
                                           int maxConcurrency,
                                           Consumer<List<T>> taskConsumer, String threadPoolName) {

        if (list.isEmpty()) {
            return;
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.start:{}", threadPoolName);
        Semaphore semaphore = new Semaphore(maxConcurrency);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Future<Boolean>> futures = new ArrayList<>();
        for (List<T> ts : list) {
            if (isException.get()) {
                throw new java.lang.RuntimeException("queryTransformUpdateAsync.isException.true");
            }
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                isException.set(true);
                LOGGER.error("获取许可异常", e);
                throw new java.lang.RuntimeException("executeConcurrentTasks.list.semaphore.acquire.InterruptedException", e);
            }
            futures.add(threadPoolService.submit(() -> {
                try {
                    I18nMessageUtils.initThreadLocale();
                    taskConsumer.accept(ts);
                    return true;
                } catch (Exception e) {
                    isException.set(true);
                    LOGGER.error("Async task failed", e);
                    throw e;
                } finally {
                    I18nMessageUtils.remove();
                    semaphore.release();
                }
            }));
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.wait{}", futures.size());
        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (Exception e) {
                throw new java.lang.RuntimeException("executeConcurrentTasks.list.future.get.InterruptedException", e);
            }
        }
        if (isException.get()) {
            throw new java.lang.RuntimeException("Async execution failed");
        }
        LOGGER.info("executeConcurrentTasks.threadPoolName.end:{}", threadPoolName);
    }


    /**
     * 异步执行任务 每次处理一条数据
     */
    public <T> void executeConcurrentTask(ExecutorService threadPoolService,
                                          Collection<T> list,
                                          int maxConcurrency,
                                          Consumer<T> taskConsumer, String threadPoolName) {

        if (list.isEmpty()) {
            return;
        }
        LOGGER.info("executeConcurrentTask.threadPoolName.start:{}", threadPoolName);
        Semaphore semaphore = new Semaphore(maxConcurrency);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Future<Boolean>> futures = new ArrayList<>();
        for (T ts : list) {
            if (isException.get()) {
                throw new java.lang.RuntimeException("queryTransformUpdateAsync.isException.true");
            }
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                isException.set(true);
                LOGGER.error("获取许可异常", e);
                throw new java.lang.RuntimeException("executeConcurrentTasks.list.semaphore.acquire.InterruptedException", e);
            }
            futures.add(threadPoolService.submit(() -> {
                try {
                    I18nMessageUtils.initThreadLocale();
                    taskConsumer.accept(ts);
                    return true;
                } catch (Exception e) {
                    isException.set(true);
                    LOGGER.error("Async task failed", e);
                    throw e;
                } finally {
                    I18nMessageUtils.remove();
                    semaphore.release();
                }
            }));
        }
        LOGGER.info("executeConcurrentTask.threadPoolName.wait{}", futures.size());
        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (Exception e) {
                throw new java.lang.RuntimeException("executeConcurrentTasks.list.future.get.InterruptedException", e);
            }
        }
        if (isException.get()) {
            throw new java.lang.RuntimeException("Async execution failed");
        }
        LOGGER.info("executeConcurrentTask.threadPoolName.end:{}", threadPoolName);
    }


    /**
     * 异步执行任务 每次处理一条数据，并返回  结果列表
     */
    public <T, R> List<R> executeConcurrentTaskWithResult(ExecutorService threadPoolService,
                                                          Collection<T> list,
                                                          int maxConcurrency,
                                                          Function<T, R> taskFunction, String threadPoolName) {
        if (list.isEmpty()) {
            return Collections.emptyList();
        }

        LOGGER.info("executeConcurrentTaskWithResult.threadPoolName.start:{}", threadPoolName);
        Semaphore semaphore = new Semaphore(maxConcurrency);
        AtomicBoolean isException = new AtomicBoolean(false);
        List<Future<R>> futures = new ArrayList<>();

        for (T ts : list) {
            if (isException.get()) {
                throw new RuntimeException("queryTransformUpdateAsync.isException.true");
            }

            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                isException.set(true);
                LOGGER.error("获取许可异常", e);
                throw new RuntimeException("executeConcurrentTasks.list.semaphore.acquire.InterruptedException", e);
            }

            // 提交 Callable 以获取返回值
            Future<R> future = threadPoolService.submit(() -> {
                try {
                    I18nMessageUtils.initThreadLocale();
                    return taskFunction.apply(ts); // 执行任务并返回结果
                } catch (Exception e) {
                    isException.set(true);
                    LOGGER.error("Async task failed", e);
                    throw e;
                } finally {
                    I18nMessageUtils.remove();
                    semaphore.release();
                }
            });

            futures.add(future);
        }

        LOGGER.info("executeConcurrentTaskWithResult.threadPoolName.wait {}", futures.size());
        List<R> results = new ArrayList<>();
        // 可选：等待所有任务完成并检查异常
        for (Future<R> future : futures) {
            try {
                results.add(future.get()); // 等待任务完成
            } catch (ExecutionException | InterruptedException e) {
                throw new RuntimeException("executeConcurrentTasks.list.future.get.exception", e);
            }
        }

        if (isException.get()) {
            throw new RuntimeException("Async execution failed");
        }
        LOGGER.info("executeConcurrentTaskWithResult.threadPoolName.end:{}", threadPoolName);
        return results;
    }

}
